package com.lagou.kafka.demo.producer;

import com.lagou.kafka.demo.entity.User;
import com.lagou.kafka.demo.serializer.UserSerializer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @param
 * @author oyz
 * @version 1.0.4
 * @description: 发送者
 * @date 2022/8/17 21:37
 */
public class MyProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String, Object> producerMap = new HashMap<>(3);
        producerMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.106.130:9092");
        // 配置key的序列化器
        producerMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 配置自定义的value序列化器
        producerMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserSerializer.class);
        User user = new User();
        user.setUserName("oyz");
        user.setUserId(1);
        ProducerRecord<String, User> producerRecord = new ProducerRecord<>("tp_user_01", user.getUserName(),user);
        KafkaProducer<String,User>   producer = new KafkaProducer<>(producerMap);
        // 消息的同步确认的方式
        /*final Future<RecordMetadata> future = producer.send(producerRecord);
        final RecordMetadata recordMetadata = future.get();
        System.out.println("消息的主题："+recordMetadata.topic());
        System.out.println("消息的分区号："+recordMetadata.partition());
        System.out.println("消息的偏移量："+recordMetadata.offset());*/

        producer.send(producerRecord, (metadata, exception) -> {
            if(exception != null){
                throw new RuntimeException("消息发送异常:"+exception.getMessage());
            }else{
                System.out.println("消息的主题："+metadata.topic());
                System.out.println("消息的分区号："+metadata.partition());
                System.out.println("消息的偏移量："+metadata.offset());
            }
        });
        // 关闭生产者
        producer.close();


    }

}
